package com.example.mq.mqserver;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.*;
import com.example.mq.mqserver.datacenter.DiskDataCenter;
import com.example.mq.mqserver.datacenter.MemoryDataCenter;
import jdk.nashorn.internal.ir.BaseNode;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Getter
//通过这个类表示虚拟主机
//作为业务逻辑的整合 就需要对代码中抛出的异常进行处理了
public class VirtualHost {
    private String virtualHostName;
    //内存中数据和硬盘中数据存储
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    //匹配规则
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);
    //队列锁和交换机锁
    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;
        //对于memoryDataCenter不需要初始化 在类内部已经初始化过了
        //对于diskDataCenter来说 要进行初始化操作
        diskDataCenter.init();
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            log.info("恢复内存数据失败");
        }
    }

    //创建交换机
    //如果交换机不存在 则创建 如果存在 直接返回
    //返回值是boolean 创建成功true 失败 返回false
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete,
                                   Map<String, Object> arguments) {
        //把交换机的名字 加上虚拟主机作为前缀
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    //该交换机已经存在
                    log.info("交换机已经存在 exchangeName = " + exchangeName);
                    return true;
                }
                //真正创建交换机
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);

                //把交换机对象写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //4.把交换机对象写入内存
                memoryDataCenter.insertExchange(exchange);
                log.info("交换机创建完成 exchangeName = " + exchangeName);
                //先硬盘 后内存 因为内存容易失败 如果硬盘失败 就不向内存中存储
            }
            return true;
        } catch (Exception e) {
            log.info("交换机创建失败 exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                //先找到对应的交换机
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MqException("交换机不存在 无法删除");
                }
                //删除硬盘上的数据
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                //删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                log.info("交换机删除成功 exchangeName = " + exchangeName);

            }
            return true;
        } catch (Exception e) {
            log.info("交换机删除失败 exchangeName = " + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete,
                                Map<String, Object> arguments) {
        //拼接队列名字
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    log.info("队列已经存在 queueName = " + queueName);
                    return true;
                }
                //创建队列对象
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                //写硬盘
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                //写内存
                memoryDataCenter.insertQueue(queue);
                log.info("队列创建成功 queueName = " + queueName);
            }
            return true;
        } catch (Exception e) {
            log.info("队列创建失败 queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除队列
    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if (queue == null) {
                throw new MqException("队列不存在 无法删除 queueName = " + queueName);
            }
            //删除硬盘数据
            if (queue.isDurable()) {
                diskDataCenter.deleteQueue(queueName);
            }
            memoryDataCenter.deleteQueue(queueName);
            log.info("交换机删除成功 exchangeName = " + queueName);
            return true;
        } catch (Exception e) {
            log.info("队列创建失败 + queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }
    //绑定
    public boolean queueBind(String queueName, String exchangeName, String bindingKey) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding existsBinding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (existsBinding != null) {
                        throw new MqException("binding 已经存在 queueName = " + queueName);
                    }
                    //验证bindingkey是否合法
                    if (!router.checkBindingKey(bindingKey)) {
                        throw new MqException("bindingKey非法 bindingKey = " + bindingKey);
                    }
                    //创建Binding对象
                    Binding binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    //获取对应的交换机和队列 如果交换机或者队列不存在 也无法创建
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if (queue == null) {
                        throw new MqException("队列不存在 queueName = " + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if (exchange == null) {
                        throw new MqException("交换机不存在 exchangeName = " + exchangeName);
                    }
                    //先写硬盘
                    if (queue.isDurable() && exchange.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    //在写内存
                    memoryDataCenter.insertBinding(binding);
                    log.info("绑定成功 exchangeName = " + exchangeName + "queueName + " + queueName);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("绑定失败 exchangeName = " + exchangeName + "queueName + " + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueUnbind(String queueName, String exchangeName) {
        queueName = virtualHostName + queueName;
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    //获取binding看是否已经存在
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MqException("删除绑定失败 绑定不存在 exchangeName" + exchangeName + ", queueName" + queueName);
                    }
                    //无论绑定是否持久化 都进行删除
                    diskDataCenter.deleteBinding(binding);
                    //删除内存的数据
                    memoryDataCenter.deleteBinding(binding);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("删除绑定失败");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties, byte[] body) {
        try {
            //转换交换机的名字
            exchangeName = virtualHostName + exchangeName;
            //检查routingKey是否合法
            if (!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 非法 routingKey = " + routingKey);
            }
            //查找交换机对象
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if (exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在 exchangeName = " + exchangeName);
            }
            //判定交换机的类型
            if (exchange.getType() == ExchangeType.DIRECT) {
                //直接交换机的转发规则
                //以routingKey作为队列的名字 直接把消息写入指定的队列中
                //此时可以无视绑定关系
                String queueName = virtualHostName + routingKey;
                //构造消息对象
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                //查找该队列名对应的对象
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new Exception("[VirtualHost] 队列不存在 queueName = " + queueName);
                }
                //队列存在 直接给队列中写入消息
                sendMessage(queue, message);
            } else {
                //以fanout和topic的方式转发
                //找到该交换机关联的所有绑定 并遍历这些绑定对象
                ConcurrentHashMap<String, Binding> bindingsMap = memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String, Binding> entry : bindingsMap.entrySet()) {
                    //获取到绑定对象
                    Binding binding = entry.getValue();
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if (queue == null) {
                        //此处不跑出异常 可能有多个这样的队列
                        //希望不要因为一个队列的失败 影响其他队列消息的传输
                        log.info("basicPublish 发送消息时 发现消息不存在 queueName = " + binding.getQueueName());
                        continue;
                    }
                    //构造消息对象
                    Message message = Message.createMessageWithId(exchangeName, basicProperties, body);
                    //判断这个消息是否能转发给该队列
                    //如果是fanout 则所有绑定的队列都要转发
                    //如果是topic 还要判定下 bindkey和routingKey是否匹配
                    if (!router.route(exchange.getType(), binding, message)) {
                        continue;
                    }
                    //真正转发消息给队列
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            log.info("消息发送失败");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws IOException, MqException, InterruptedException {
        // 此处发送消息 就是把消息写入硬盘和内存中
        int deliverMode = message.getDeliverMode();
        //deliverMode 为1 不持久化 为2 不持久化
        if (deliverMode == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        memoryDataCenter.sendMessage(queue, message);

        //通知消费者可以消费消息了
        consumerManager.notifyConsumer(queue.getName());
    }

    //添加一个队列的订阅者 队列收到消息之后 就要把消息推送给对应的订阅者
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck, Consumer consumer) {
        //构造一个consumerEnv对象 把这个对应的队列找到 再把这个Consumer对象添加到队列中
        queueName = virtualHostName + queueName;
        try {
            consumerManager.addConsumer(consumerTag, queueName, autoAck, consumer);
            log.info("basicConsumer成功 queueName :{}", queueName);
            return true;
        } catch (Exception e) {
            log.info("basicConsumer 失败 queueName :{}", queueName);
            e.printStackTrace();
            return false;
        }
    }

    //只有在手动应答时 才调用 应答成功 删除这条消息
    public boolean basicAck(String queueName, String messageId){
        queueName = virtualHostName + queueName;
        try {
            //获取到消息和队列
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null){
                throw new MqException("[VirtualHost] 要确认的消息不存在 messageId = " +messageId);
            }
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null){
                throw new MqException("[virtualHost] 要确认的队列不存在 queueName = " + queueName);
            }
            //删除硬盘上的数据
            if(message.getDeliverMode() == 2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //删除消息中心的数据
            memoryDataCenter.removeMessage(messageId);
            //删除待确认集合中的消息
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            log.info("basicAck成功 消息被成功确认 queueName:{},messageId:{}",queueName,messageId);
            return true;
        } catch (MqException | IOException | ClassNotFoundException e) {
            log.info("basicAck失败 消息确认失败 queueName:{},messageId:{}",queueName,messageId);
            return false;
        }
    }
}
